<?php
/**
 * Created by PhpStorm.
 * User: xp
 * Date: 2021/6/23
 * Time: 9:46
 */

namespace app\common\lib;


use app\exception\IndexException;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class RabbitMq
{
    protected static $instance;

    protected $config = [];    //['host'='','port'=>'','username'=>'','password'=>'']

    protected $connection;

    protected $channel;

    public function __construct()
    {
        $this->config = config('rabbitmq.connection');
    }

    public static function getInstance()
    {
//        if(is_object(self::$instance)){
//            return self::$instance;
//        }

        return self::$instance = new self();
    }

    public function getConnection()
    {
        try{
            if(!empty($this->config['host']) && !empty($this->config['port']) && !empty($this->config['username']) && !empty($this->config['password'])) {
                $this->connection = new AMQPStreamConnection($this->config['host'], $this->config['port'], $this->config['username'], $this->config['password']);
                if($this->connection==false){
                   throw new IndexException('rabbitmq 连接失败', 600);
                }

                $this->channel = $this->connection->channel();

                return [$this->connection, $this->channel];
            }
        }catch (\Throwable $e){
            throw $e;
        }
    }

    public function publicExchange($data, $exchangeName, $exchange='fanout', $routeKey='')
    {
        $this->getConnection();

        $this->channel->exchange_declare($exchangeName, $exchange, false, true, false);

        if(is_array($data)){
            $data = json_encode($data, 256);
        }elseif(is_object($data)){
            $data = serialize($data);
        }

        $msg = new AMQPMessage(
            $data,
            ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]
        );

        $this->channel->basic_publish($msg, $exchangeName, $routeKey);

        $this->channel->close();
        $this->connection->close();

        return $this;
    }

    public function receiveExchange($callback, $exchangeName, $exchange='fanout', $routeKey='')
    {
        $this->getConnection();

        $this->channel->exchange_declare($exchangeName, $exchange, false, true, false);

        list($queue_name, ,) =  $this->channel->queue_declare("", false, false, true, false);

        $this->channel->queue_bind($queue_name, $exchangeName, $routeKey);

        $this->channel->basic_qos(null, 1, null);

        $this->channel->basic_consume($queue_name, '', false, false, false, false, $callback);

        while ( $this->channel->is_open()) {
            $this->channel->wait();
        }

        $this->channel->close();
        $this->connection->close();

        return $this;
    }

    public function publicQueue($queueName,$data)
    {
        $this->getConnection();

        $env = config('app.env','local');

        $this->channel->queue_declare($env.'_'.$queueName, false, true, false, false);

        if(is_array($data)){
            $data = json_encode($data, 256);
        }elseif(is_object($data)){
            $data = serialize($data);
        }
        $msg = new AMQPMessage(
            $data,
            ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]
        );

        $this->channel->basic_publish($msg, '', $env.'_'.$queueName);

        $this->channel->close();
        $this->connection->close();

        return $this;
    }

    public function receiveQueue($queueName,$callback)
    {
        $this->getConnection();

        if($this->queueStatus($queueName)){
            //测试和正式上只能开一个消费客户端
            return $this;
        }

        $env =  config('app.env','local');
        $this->channel->queue_declare($env.'_'.$queueName, false, true, false, false);

        $funcCallback= function ($msg) use($callback){
            $queueName = $msg->getRoutingKey();
            $content = "\n";
            $content .= '   ['.$queueName.' start] Received '. $msg->body. "\n";

            if(is_callable($callback)){
                $data = @json_decode($msg->body,true);
                $data===null && $data = @unserialize($msg->body);
                $data===false && $data = $msg->body;

                try{
                    $result = call_user_func($callback, ['data'=>$data]);
                    $content .= "   [runing] ". json_encode($result, 256)."\n";
                }catch (\Exception $e){
                    $content .= "   [exception] ". $e->getMessage()."\n";
                }
            }

            //sleep(substr_count($msg->body, '.'));

            $content .= "   [$queueName end] Done\n";

            echo $content;
            //消费者日记
            Common::createAndSaveLog($content,'rabbitmq_consumers_log');
            $msg->ack();
        };

        $this->channel->basic_qos(null, 1, null);
        $this->channel->basic_consume($env.'_'.$queueName, '', false, false, false, false, $funcCallback);

        $startTime = time();
        while ( $this->channel->is_open()) {
            //超一个小时，就关掉。
            if((time()-$startTime)>=3600){
                $this->channel->close();
                $this->connection->close();

                $this->closeQueue($queueName);
            }
            $log = [
                'time'=>date('Y-m-d H:i:s'),
                'status'=>$this->channel->is_open(),
            ];
            file_put_contents(runtime_path('rebbitmq_queue.log'),json_encode($log,256));

            $this->channel->wait();
        }

        $this->closeQueue($queueName);

        $this->channel->close();

        $this->connection->close();

        return $this;
    }

    public function closeQueue($queueName)
    {
        $env =  config('app.env','local');
        $output = false;
        if($env!='local'){
            exec('ps -ef|grep "rabbitmq:queue '.$queueName.'"|grep -v grep|cut -c 9-15|xargs kill -9',$output);
            if(count($output)>1 && isset($output[0])){
                exec('kill -9 '.trim($output[0]));
            }
        }

        return $output;
    }

    public function queueStatus($queueName)
    {
        $env =  config('app.env','local');
        $output = [];
        if($env!='local'){
            exec('ps -ef|grep "rabbitmq:queue '.$queueName.'"|grep -v grep|cut -c 9-15',$output,$status);
        }

        return count($output)>1 ? true : false;
    }

    public function closeAllQueue()
    {
        $env =  config('app.env','local');
        $output = [];
        if($env!='local'){
            exec('ps -ef|grep "rabbitmq:queue"|grep -v grep|cut -c 9-15|xargs kill -9',$output);
        }

        return $output;
    }
}